package com.foogui.demo.consume;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;

import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.stereotype.Service;

import java.util.List;

@Slf4j
//@Service
@RocketMQMessageListener(consumerGroup = "consume_group", topic = "topic1")
public class Consumer {

    public static void main(String[] args) throws Exception {
        // 1.定义一个push消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consume_group");
        // 2.指定Nameserver地址
        consumer.setNamesrvAddr("localhost:9876");
        // 3.订阅主题和tag，设置消费模式
        consumer.subscribe("topic1", "*");
        //默认集群消费，负载均衡
        //consumer.setMessageModel(MessageModel.CLUSTERING);
        // 指定从第一条消息开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //设置广播消费
        consumer.setMessageModel(MessageModel.BROADCASTING);
        // 4.设置回调函数，处理消息

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                // 逐条消费消息
                for (MessageExt msg : list) {
                    log.info("consumeThread = " + Thread.currentThread().getName() + ", msg = " + new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 5.启动消费者
        consumer.start();
    }
}
